package org.apache.kylin.engine.streaming.util;

import org.apache.commons.lang.StringUtils;
import org.apache.kylin.common.streaming.StreamingBatch;
import org.apache.kylin.common.util.ClassUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileFilter;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

/**
 * Created by taoliu on 2016/12/14.
 */
public class SparkEnv {
    private static final Logger logger = LoggerFactory.getLogger(SparkEnv.class);
    private static JavaSparkContext context = null;

    public static JavaSparkContext getContext(){
        if (context == null){
            init();
        }
        return context;
    }

    public static void init(){
        SparkConf conf = new SparkConf();
        conf.set("spark.storage.memoryFraction", "0.3");
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
        context = new JavaSparkContext(conf);
    }

    public static void close(){
        if (context != null){
            context.close();
        }
    }

    public static void setConf2Classpath(String confPath){
        if(StringUtils.isEmpty(confPath)){
            confPath = System.getenv().get("KYLIN_HOME");
            if (StringUtils.isEmpty(confPath)){
                logger.warn("KYLIN_HOME not set and spark build conf path is not special");
                return;
            }
            confPath = confPath + "/conf";
        }
        ClassUtil.addClasspath(confPath);
    }

    private static Set<String> kylinConfigFiles = new HashSet<String>();
    static {
        kylinConfigFiles.add("sparkConf.xml");
        kylinConfigFiles.add("kylin.properties");
    }
    public static void addConf2Context(JavaSparkContext sc,String confPath){
        final File[] files = new File(confPath).listFiles(new FileFilter() {
            @Override
            public boolean accept(File pathname) {
                if (pathname.getAbsolutePath().endsWith(".xml") && kylinConfigFiles.contains(pathname.getName())) {
                    return true;
                }
                if (pathname.getAbsolutePath().endsWith(".properties") && kylinConfigFiles.contains(pathname.getName())) {
                    return true;
                }
                return false;
            }
        });
        if (files == null){
            return;
        }
        for (File file : files) {
            sc.addFile(file.getAbsolutePath());
        }
    }
}
